#pragma once

#include <unordered_set>
#include <exception>

#include "messaging/drivers/drivers.h"
#include "messaging/drivers/hiredis/async.h"
#include "event/event.h"
#include "common/queue.h"
#include "common/format.h"
#include "common/uuid.h"
#include "common/str.h"
#include "common/log.h"


namespace evm { namespace messaging {

using namespace std;
using namespace event;
using common::Queue;

// Polling timeout in milliseconds.
static const int POLL_TIMEOUT = 100;

struct RedisAsync {
    enum {
        CONNECTED, DISCONNECTED, ERROR
    };

    RedisAsync(const string& name, const string& ip, int port,
               EventLoopPtr loop, size_t retry_time = 1);

    void start(bool wait = true);

    void stop(bool wait = true);

    void connect();

    void disconnect();

    /* Async connect after `retry_time` seconds. */
    void reconnect();

    void change_state(int state) {
        this->state = state;
    }

    bool command(string cmd);

    bool command(const char *format, ...);

    virtual void rcallback(redisReply*) = 0;

private:
    string name;
    string ip;
    int port;
    EventLoopPtr loop;
    size_t retry_time;

    redisAsyncContext* ctx;
    int fd;

    /* Whether read event or write event has been added to the event loop. */
    bool reading{}, writing{};
    recursive_mutex rw_mutex;  // mutex for reading or writing async context

    atomic<int> state{DISCONNECTED};
    recursive_mutex conn_mutex;  // mutex for starting or stopping connection
    condition_variable_any conn_cv;  // condition variable for starting or stopping connection

    FileHandler read_handler;
    FileHandler write_handler;

    static void replied(redisAsyncContext *c, void *r, void *privdata);

    static void connected(const redisAsyncContext* ctx, int status);

    static void disconnected(const redisAsyncContext* ctx, int status);

    static void add_read(void* data);

    static void del_read(void* data);

    static void add_write(void* data);

    static void del_write(void* data);

    static void cleanup(void* data);
};

using RedisAsyncPtr = shared_ptr<RedisAsync>;

inline string to_str(redisReply* r) {
    return string(r->str, static_cast<size_t>(r->len));
}

struct Event {
    Json wait(int timeout = POLL_TIMEOUT);

    void notify(Json&& msg);

    Json msg;
    mutex m;
    condition_variable cv;
};

using EventPtr = shared_ptr<Event>;

struct RedisSender : RedisAsync {
    RedisSender(const string& name, const string& ip, int port, EventLoopPtr loop)
        : RedisAsync(name, ip, port, loop) {}

    bool publish(string&& channel, Json&& msg);

    bool push(string&& list, Json&& msg);

    void rcallback(redisReply* r);

    unordered_set<string> channels;
    unordered_set<string> lists;
};

using RedisSenderPtr = shared_ptr<RedisSender>;

using MsgHandler = function<void (const string&, const string&)>;

struct RedisSubscriber : RedisAsync {
    RedisSubscriber(const string& name, const string& ip, int port, EventLoopPtr loop)
        : RedisAsync(name, ip, port, loop) {
    }

    void preset(vector<string>&& channels, MsgHandler handler);

    void subscribe() {
        if (!command(cmd_str))  // TODO: should send binary safe string
            log_error("redis command fail");
    }

    void rcallback(redisReply* r);

    MsgHandler handler;
    unordered_set<string> channels;
    string cmd_str;
};

using RedisSubscriberPtr = shared_ptr<RedisSubscriber>;

struct RedisPoper : RedisAsync {
    RedisPoper(const string& name, const string& ip, int port, EventLoopPtr loop)
        : RedisAsync(name, ip, port, loop) {}

    void preset(unordered_set<string>&& lists, int timeout = -1, MsgHandler handler = nullptr);

    void pop() {
        if (!command(cmd_str))
            log_error("redis command fail");
    }

    void rcallback(redisReply* r);

    MsgHandler handler;
    unordered_set<string> lists;
    string cmd_str;
    bool in_poping{};
};

using RedisPoperPtr = shared_ptr<RedisPoper>;

struct IncomingMsg : BaseIncomingMsg {
    using Handler = function<void (Json&&)>;

    IncomingMsg(Json&& message = Json(), Handler handler = nullptr)
        : BaseIncomingMsg(std::move(message)), handler(handler) {}

    /* Send a reply or failure back to the client. */
    void reply(Json&& msg) {
        if (handler)
            handler({{"reply", std::move(msg)}});
    }

    void reply_error(Json&& msg) {
        if (handler)
            handler({{"error", std::move(msg)}});
    }

    Handler handler;
};

struct RPCListener: BaseListener {
    RPCListener(TargetPtr target, const string& ip, int port, EventLoopPtr loop);

    void start() {
        BaseListener::start();

        poper.start();
        subscriber.start();
        sender.start();

        subscriber.subscribe();
        if (!target->fanout)
            poper.pop();
    }

    void stop() {
        poper.stop();
        subscriber.stop();
        sender.stop();

        queue.clear();

        BaseListener::stop();
    }

    IncomingMsgPtr poll();

    TargetPtr target;

    Queue<Json> queue;

    // For a specified target, only one of these two listeners will poll.
    RedisPoper poper;
    RedisSubscriber subscriber;

    // Message replier.
    RedisSender sender;
};

struct NotificationListener: BaseListener {
    NotificationListener(const vector<TargetPriority>& targets_and_priorities,
                         const string& ip, int port, EventLoopPtr loop);

    void start() {
        BaseListener::start();
        subscriber.start();
        subscriber.subscribe();
    }

    void stop() {
        subscriber.stop();
        queue.clear();
        BaseListener::stop();
    }

    IncomingMsgPtr poll();

    TargetPtr target;
    Queue<Json> queue;
    RedisSubscriber subscriber;
};

struct Sender {
    Sender(const string& ip, int port, EventLoopPtr loop);

    void start();

    void stop();

    bool publish(string channel, Json&& msg) {
        return sender.publish(std::move(channel), std::move(msg));
    }

    bool push(string list, Json msg) {
        return sender.push(std::move(list), std::move(msg));
    }

    Json send(TargetPtr target, const Json& msg, bool wait_for_reply = false,
              int timeout = -1, int retry = -1);

    RedisSender sender;
    RedisPoper poper;
    // For message needing reply, receiver should reply to the list named `reply_id`.
    string reply_id;
    unordered_map<string, EventPtr> events;
    mutex events_mutex;
    atomic<bool> stopped{true};
};

using SenderPtr = shared_ptr<Sender>;

struct RedisDriver : BaseDriver {
    RedisDriver(const string& ip, int port)
        : ip(ip), port(port), loop(make_shared<EventLoop>(64)),
        sender(ip, port, loop){
    }

    void start();

    void stop();

    Json send(TargetPtr target, Json&& msg,
              bool wait_for_reply = false, int timeout = -1, int retry = -1) {
        return sender.send(target, std::move(msg), wait_for_reply, timeout, retry);
    }

    void send_notification(TargetPtr target, Json&& msg, int retry = -1) {
        sender.publish(target->topic, std::move(msg));
        (void)retry;
    }

    ListenerPtr listen(TargetPtr target) {
        auto listener = ListenerPtr(new RPCListener(target, ip, port, loop));
        listener->start();
        listeners.push_back(listener);
        return listener;
    }

    ListenerPtr listen_for_notifications(const vector<TargetPriority>& targets_and_priorities) {
        auto listener = ListenerPtr(new NotificationListener(targets_and_priorities, ip, port, loop));
        listener->start();
        listeners.push_back(listener);
        return listener;
    }

private:
    const string ip;
    int port;
    EventLoopPtr loop;

    Sender sender;

    vector<ListenerPtr> listeners;
};

}}
